package com.example.websocketclient.websocket;

import com.alibaba.fastjson.JSONObject;
import com.rabbitmq.client.*;
import org.java_websocket.client.WebSocketClient;
import org.java_websocket.handshake.ServerHandshake;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.net.URI;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Map;
import java.util.concurrent.*;

@Component
public class CustomWebSocketClien extends WebSocketClient {
    private static final ScheduledExecutorService executor = Executors.newScheduledThreadPool(1);
    private static final int RECONNECT_INTERVAL = 5; // 重连间隔（秒）
    private static final int MAX_RETRY_COUNT = 10;// 最大重连次数
    private int retryCount = 0;// 当前重连次数
    private final static String QUEUE_NAME = "test_queue";

    // 添加配置注入
    @Value("${rabbitmq.host}")
    private String rabbitmqHost;

    @Value("${rabbitmq.port}")
    private int rabbitmqPort;

    @Value("${rabbitmq.username}")
    private String rabbitmqUsername;

    @Value("${rabbitmq.password}")
    private String rabbitmqPassword;

    public CustomWebSocketClien(URI serverUri) {
        super(serverUri);
    }

    private void sendMessagesToRabbitMQ(String message) {
        try {
            // 连接到RabbitMQ
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost(rabbitmqHost);
            factory.setPort(rabbitmqPort);
            factory.setUsername(rabbitmqUsername);
            factory.setPassword(rabbitmqPassword);

            // 声明队列并发送消息
            try (Connection connection = factory.newConnection(); Channel channel = connection.createChannel()) {
                channel.queueDeclare(QUEUE_NAME, true, false, false, null);
                //队列和消息均持久化
                AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
                        .deliveryMode(2)
                        .build();
                // Topic交换机测试示例
                channel.exchangeDeclare("test_topic", BuiltinExchangeType.TOPIC);
                channel.queueBind(QUEUE_NAME, "test_topic", "user.*");
                channel.basicPublish("test_topic", "user.create", props, message.getBytes());
            } catch (IOException | TimeoutException e) {
                throw new RuntimeException(e);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public void connect() {
        super.connect();
    }

    // 连接成功回调
    @Override
    public void onOpen(ServerHandshake handshakedata) {
        System.out.println("连接成功");

//        sendMessage("客户端注册:设备001"); // 发送自定义初始化消息
//        startHeartbeat(); // 启动心跳
    }

    // 使用ConcurrentHashMap记录消息类型最后处理时间
    private static final Map<String, Long> MSG_CACHE = new ConcurrentHashMap<>();

    // 消息接收回调
    @Override
    public void onMessage(String message) {
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        String formattedDate = sdf.format(new Date());
//        System.out.println("收到消息" + formattedDate + ":" + message);

        JSONObject jsonObject = JSONObject.parseObject(message);
        // 处理消息
        if ("notify_history_changed".equals(jsonObject.getString("method"))) {
            // 特殊消息实时保存
            sendMessagesToRabbitMQ(message);
            System.out.println("保存消息" + formattedDate + ":" + message);
        } else if ("notify_status_update".equals(jsonObject.getString("method"))) {
            String msgType = jsonObject.getString("method");
            long currentTime = System.currentTimeMillis();

            // 检查时间窗口
            if (MSG_CACHE.containsKey(msgType)) {
                long lastTime = MSG_CACHE.get(msgType);
                if (currentTime - lastTime < 5 * 1000L) {
                    return; // N秒内同类型消息丢弃
                }
            }

            // 更新缓存并处理消息
            MSG_CACHE.put(msgType, currentTime);
            sendMessagesToRabbitMQ(message);
            System.out.println("保存消息" + formattedDate + ":" + message);
        }
    }

    // 定时清理过期缓存（防止内存泄漏）
    @Scheduled(fixedRate = 60000)
    public void cleanCache() {
        MSG_CACHE.entrySet().removeIf(entry ->
                System.currentTimeMillis() - entry.getValue() > 5 * 1000L
        );
    }

    // 连接关闭回调（含自动重连）
    @Override
    public void onClose(int code, String reason, boolean remote) {
        System.out.println("连接关闭 code:" + code + " reason:" + reason);
        scheduleReconnect();
    }

    // 错误处理
    @Override
    public void onError(Exception ex) {
        ex.printStackTrace();
        this.close();  // 触发onClose重连
    }

    // 发送自定义消息（线程安全）
    public synchronized void sendMessage(String message) {
        if (this.isOpen()) {
            try {
                send(message);
            } catch (Exception e) {
                System.err.println("发送失败: " + e.getMessage());
            }
        }
    }

    // 心跳机制
    private void startHeartbeat() {
        executor.scheduleAtFixedRate(() -> {
            if (this.isOpen()) {
                sendMessage("ping-" + System.currentTimeMillis());
            }
        }, 10, 30, TimeUnit.SECONDS); // 30秒一次心跳
    }

    // 自动重连逻辑
    private synchronized void scheduleReconnect() {
        if (!this.isOpen() && retryCount < MAX_RETRY_COUNT) {
            retryCount++;
            executor.schedule(() -> {
                try {
                    System.out.println("尝试重新连接...(" + retryCount + "/" + MAX_RETRY_COUNT + ")");
                    this.reconnectBlocking();  // 阻塞式重连
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }, RECONNECT_INTERVAL, TimeUnit.SECONDS);
        } else if (retryCount >= MAX_RETRY_COUNT) {
            System.out.println("已达到最大重连次数，停止重连");
        }
    }

    //主动关闭连接
    public synchronized void shutdown() {
        try {
            // 关闭连接（新增关闭逻辑）
            if (isOpen()) {
                close(1000, "用户发起的关闭");

                System.out.println("连接已主动关闭");

                // 立即关闭线程池
                executor.shutdownNow();
                // 清除所有预定任务
                MSG_CACHE.clear();
                // 关闭线程池（新增资源释放）
                executor.shutdown();
                if (!executor.awaitTermination(3, TimeUnit.SECONDS)) {
                    executor.shutdownNow();
                }
            }
        } catch (Exception e) {
            System.err.println("关闭异常:" + e.getMessage());
        }
    }
}
